/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * Author: JArod Wen
 * Created: May 4, 2011
 * 
 */
package edu.ucr.cs.dblab.acestore.operators;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.nio.ByteBuffer;

import edu.ucr.cs.dblab.acestore.api.IOperator;
import edu.ucr.cs.dblab.acestore.api.ISinglePrevOperatorFactory;
import edu.ucr.cs.dblab.acestore.api.TupleInterpreter;
import edu.ucr.cs.dblab.acestore.base.ACEContext;
import edu.ucr.cs.dblab.acestore.base.datatypes.ISerDeserializer;
import edu.ucr.cs.dblab.acestore.base.datatypes.IntegerSerDeserializer;
import edu.ucr.cs.dblab.acestore.base.datatypes.SchemaDescriptor;

/**
 * @author wenj
 * 
 */
public class FilePlainOutputOperatorFactory implements ISinglePrevOperatorFactory {

    private final SchemaDescriptor outputSchemaDescriptor;

    private final String delimiter;

    private final String filepath;

    public FilePlainOutputOperatorFactory(String filepath,
            SchemaDescriptor outputSchemaDescriptor, String delimiter) {
        this.outputSchemaDescriptor = outputSchemaDescriptor;
        this.filepath = filepath;
        this.delimiter = delimiter;
    }

    @Override
    public IOperator createOperator(ACEContext ctx, final IOperator prevOperator) {

        return new IOperator() {

            BufferedWriter bwriter;
            TupleInterpreter tupleInterpreter;

            @Override
            public void open() throws Exception {
                bwriter = new BufferedWriter(new FileWriter(filepath));
                prevOperator.open();
            }

            @Override
            public void getNext(ByteBuffer page) throws Exception {
                
                while (true) {
                    prevOperator.getNext(page);
                    // Write output
                    // - Get the count of records in the page
                    int recCount = page.getInt(page.capacity() - 2 * INTSIZE - RESERVED_END);
                    int lastOffset = -1;
                    // - Write each record in the page through the output writer
                    for (int i = 0; i < recCount; i++) {
                        int len = page.getInt(page.capacity() - 2 * INTSIZE
                                - (i + 1) * INTSIZE - RESERVED_END)
                                - lastOffset;
                        tupleInterpreter = new TupleInterpreter(page.array(),
                                lastOffset + 1, len);
                        for (int j = 0; j < tupleInterpreter.getFieldCount(); j++) {
                            byte[] data = tupleInterpreter.getField(j);
                            bwriter.write(outputSchemaDescriptor.getFields()[j]
                                    .deserialize(data, 0, data.length)
                                    + delimiter);
                        }
                        bwriter.write("\n");
                        lastOffset += len;
                    }
                    // - Stop if this is the last page
                    if (page.getInt(page.capacity() - INTSIZE - RESERVED_END) < 0)
                        break;
                    else {
                        // - otherwise, load another page
                        page.position(0);
                    }

                }
            }

            @Override
            public void close() throws Exception {
                prevOperator.close();
                bwriter.close();
            }
        };
    }

    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        FileScanOperatorFactory scanOperatorFactory = new FileScanOperatorFactory(
                "/Users/jarodwen/Desktop/hyracks/data/simple2D.txt",
                new SchemaDescriptor(new ISerDeserializer[] {
                        IntegerSerDeserializer.getSingleton(),
                        IntegerSerDeserializer.getSingleton() }), ",");
        IOperator scanOperator = scanOperatorFactory.createOperator(null, null);
        FilePlainOutputOperatorFactory printOperatorFactory = new FilePlainOutputOperatorFactory(
                "/Users/jarodwen/Desktop/hyracks/data/simple2D_n.txt",
                new SchemaDescriptor(new ISerDeserializer[] {
                        IntegerSerDeserializer.getSingleton(),
                        IntegerSerDeserializer.getSingleton() }), "\t");
        IOperator printOperator = printOperatorFactory.createOperator(null,
                scanOperator);
        ByteBuffer page = ByteBuffer.wrap(new byte[100]);
        page.position(0);
        printOperator.open();
        printOperator.getNext(page);
        while (page.getInt(page.capacity() - INTSIZE - RESERVED_END) >= 0) {
            page.position(0);
            printOperator.getNext(page);
        }
        printOperator.close();
    }

}
